package com.atguigu.flink.chapter08_exec2;

import com.atguigu.flink.pojo.MyUtil;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/10/29
 *
 *      使用PV的数据，根据user_id进行去重统计。
 *          统计每小时的UV。
 *
 *
 *      增量:  无法实现
 *      全量:  process
 *                  在Process的算子中，声明一个容器(状态实现)，将过去1h所有的user_id收集起来，最好还能去重。
 *
 *                  使用MapState属于键控状态，必须KeyBy。
 *
 *    ----------------------------
 *      窗口[2017-11-26 09:00:00,2017-11-26 10:00:00) UV:28196
 *        ......
 *      窗口[2017-11-26 18:00:00,2017-11-26 19:00:00) UV:215662
 *
 *      原因： 同一个Process的并行度，为某个用户声明一个MapState。
 *              按照窗口，分时段统计。同一个用户可能在不同的时段，进行访问。
 *              这些不同时段的数据，分布在了不同的窗口中，但是在同一个并行度中，由于Key是一样的，共用同一个状态，造成累加的效果。
 *
 *              user_id=1:   8:00-9:00   访问 1 ：
 *              user_id=2:   8:00-9:00   访问 1 ：
 *              user_id=3:   8:00-9:00   访问 1 ：
 *              user_id=4:   8:00-9:00   访问 1 ：
 *                      { user_id=1:'a' , user_id=2:'a',user_id=3:'a',user_id=4:'a'  }
 *                              输出:  8:00-9:00 :  4
 *
 *
 *              user_id=1:   9:00-10:00   访问 1
 *                       { user_id=1:'a' , user_id=2:'a',user_id=3:'a',user_id=4:'a'  }
 *                           输出:  9:00-10:00 :  4
 *

 窗口[2017-11-26 09:00:00,2017-11-26 10:00:00) UV:28196
 窗口[2017-11-26 10:00:00,2017-11-26 11:00:00) UV:32160
 窗口[2017-11-26 11:00:00,2017-11-26 12:00:00) UV:32233
 窗口[2017-11-26 12:00:00,2017-11-26 13:00:00) UV:30615
 窗口[2017-11-26 13:00:00,2017-11-26 14:00:00) UV:32747
 窗口[2017-11-26 14:00:00,2017-11-26 15:00:00) UV:33898
 窗口[2017-11-26 15:00:00,2017-11-26 16:00:00) UV:34631
 窗口[2017-11-26 16:00:00,2017-11-26 17:00:00) UV:34746
 窗口[2017-11-26 17:00:00,2017-11-26 18:00:00) UV:32356
 窗口[2017-11-26 18:00:00,2017-11-26 19:00:00) UV:13
 */
public class Demo2_UV
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((u, ts) -> u.getTimestamp());

        env.readTextFile("data/UserBehavior.csv").setParallelism(1)
           .map(new MapFunction<String, UserBehavior>()
           {
               @Override
               public UserBehavior map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new UserBehavior(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       Integer.valueOf(words[2]),
                       words[3],
                       Long.valueOf(words[4]) * 1000
                   );
               }
           })
           //过滤出pv
           .filter(u -> "pv".equals(u.getBehavior()))
           .assignTimestampsAndWatermarks(watermarkStrategy)
           //.keyBy(UserBehavior::getUserId)
           .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
           .process(new ProcessAllWindowFunction<UserBehavior, String, TimeWindow>()
           {

               //去重     一个并行度的一个Key，有一个Map，还是所有Key有一个Map?
               //  在windowAll中，keyBy是否无所谓，会将所有key的数据都发往下游的一个并行度，这个并行度会创建一个MapState，所有key共用！
               //  好比类似算子状态(一个并行度一个状态，所有数据共用)。
               private MapState<Long, Object> uids;

               @Override
               public void open(Configuration parameters) throws Exception {
                   uids = getRuntimeContext().getMapState(new MapStateDescriptor<Long, Object>("uids", Long.class, Object.class));
               }

               @Override
               public void process(Context context, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {

                   //清空状态
                   uids.clear();

                   String windowSts = MyUtil.printTimeWindow(context.window());

                   for (UserBehavior userBehavior : elements) {
                       uids.put(userBehavior.getUserId(),"a");
                   }

                   //获取去重后的uid的数量
                   out.collect(windowSts + " UV:"+MyUtil.toList(uids.keys()).size());

               }
           })
           .print().setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
